package com.shujia.tour

import com.shujia.util.Geography
import org.apache.spark.internal.Logging
import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

object ProvinceTouristApp extends Logging {
  def main(args: Array[String]): Unit = {

    /**
      *
      * 省游客计算
      * 1、判断最大的出游距离是否大于10KM
      * 2、判断总的停留时间是否大于3个小时
      *
      *
      * 需要管理用户画像表获取常住地的网格
      *
      */


    if (args.length == 0) {
      log.error("输出参数为空")
      return
    }

    val day_id = args(0)

    log.info(s"当前天时间分区为：$day_id")

    val month_id: String = day_id.substring(0, 6)

    log.info(s"当前月时间分区为：$month_id")

    val spark: SparkSession = SparkSession.builder()
      .appName("ProvinceTouristApp")
      .config("spark.sql.shuffle.partitions", "20")
      .enableHiveSupport()
      .getOrCreate()

    import org.apache.spark.sql.functions._
    import spark.implicits._


    //读取停留表
    val stayPoint: DataFrame = spark.sql(s"select mdn,grid_id,county_id,duration from dwi.dwi_staypoint_msk_d where day_id=$day_id")

    //读取用户画像表
    val userTag: DataFrame = spark.sql(s"select mdn,resi_grid_id,resi_county_id from dim.dim_usertag_msk_m where  month_id=$month_id")

    //读取行政区配置表
    val adminCode: DataFrame = spark.sql("select county_id,prov_id from dim.dim_admincode")


    //注册计算网格点距离的自定义函数
    spark.udf.register("calculateLength", (resi_grid_id: String, city_id: String) => {
      //计算网格距离
      Geography.calculateLength(resi_grid_id.toLong, city_id.toLong)
    })


    /**
      *
      * 2、判断总的停留时间是否大于3个小时
      *
      * 判断游客在省内的停留时间是否大于3小时
      *
      */

    stayPoint
      .join(adminCode.hint("broadcast"), "county_id")
      //计算一个人在一个城市中总的停留时间
      .select($"mdn", $"prov_id", $"grid_id", sum($"duration") over Window.partitionBy($"mdn", $"prov_id") as "sumDuration")
      //过滤停留时间大于3小时的数据
      .where($"sumDuration" > 180)

      //1、判断最大的出游距离是否大于10KM
      .join(userTag, "mdn")
      //计算常住地网格到听停留点网格的距离
      .select($"mdn", $"prov_id", $"grid_id", $"resi_county_id", $"sumDuration", expr("calculateLength(grid_id,resi_grid_id)") as "distance")
      //计算最远距离
      .select($"mdn", $"prov_id", $"grid_id", $"resi_county_id", $"sumDuration", max("distance") over Window.partitionBy($"mdn", $"prov_id") as "d_max_distance")
      //过滤最远距离大于10Km的游客
      .where($"d_max_distance" > 10000)

      //整理数据
      .select($"mdn", $"resi_county_id" as "source_county_id", $"prov_id", round($"sumDuration" / 60, 2) as "d_stay_time", round($"d_max_distance" / 1000, 2) as "d_max_distance")
      //去重
      .distinct()

      //保存数据
      .write
      .format("csv")
      .option("sep", "\t")
      .mode(SaveMode.Overwrite)
      .save(s"/daas/motl/dal_tour/dal_tour_province_tourist_msk_d/day_id=$day_id")



    //增加分区
    spark.sql(s"alter table dal_tour.dal_tour_province_tourist_msk_d add if not exists partition(day_id='$day_id')")


  }
}
